package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.constants.ScheduleConstants;
import com.heima.common.redis.RedisCacheService;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Set;

/**
 * @author Gareth Cao
 * @date 2022/6/18 21:15
 */
@Service
@Slf4j
public class TaskServiceImpl implements TaskService {

    @Autowired
    private RedisCacheService redisCacheService;

    @Override
    public Long addTask(Task task) {
        //将任务存入到数据库
        Boolean flag = addTaskToDB(task);
        if (flag) {
            //将任务存入到redis中
            addTaskToRedis(task);
        }
        //返回任务id
        return task.getTaskId();
    }


    /**
     * 抽取的第二个方法
     *
     * @param task
     */
    private void addTaskToRedis(Task task) {
        //1 生成将任务存入redis当中对应的key
        String key = task.getTaskType() + "_" + task.getPriority();
        // TODO 存入redis分两种，list和zset
        //2.1 如果执行时间小于<=当前时间，那么将任务存入list中
        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            //第一个参数是key，第二个参数是值，需要将对象转为json字符串存入redis
            redisCacheService.lLeftPush(ScheduleConstants.TOPIC + key, JSON.toJSONString(task));
        }
        //2.2 如果执行时间大于当前时间并小于当前时间加上五分钟，那么将任务存入zset中
        //2.2.1 获取五分钟之后的时间
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE, 5);
        if (task.getExecuteTime() > System.currentTimeMillis() && task.getExecuteTime() <= calendar.getTimeInMillis()) {
            //key,value, scores就是执行的时间毫秒值
            redisCacheService.zAdd(ScheduleConstants.FUTURE + key, JSON.toJSONString(task), task.getExecuteTime());
        }
    }


    @Autowired
    private TaskinfoMapper taskinfoMapper;

    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;

    /**
     * 抽取的第一个方法
     * 将任务添加到数据库
     *
     * @param task
     */
    @Transactional
    public Boolean addTaskToDB(Task task) {
        //1 将任务添加到任务表中
        Taskinfo taskInfo = new Taskinfo();
        BeanUtils.copyProperties(task, taskInfo);
        taskInfo.setExecuteTime(new Date(task.getExecuteTime()));
        taskinfoMapper.insert(taskInfo);
        //2 由于mybatisplus的特性，我们可以立即获取任务表中生成的任务id
        task.setTaskId(taskInfo.getTaskId());
        //3 添加任务日志表
        TaskinfoLogs taskInfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(taskInfo, taskInfoLogs);
        taskInfoLogs.setStatus(ScheduleConstants.SCHEDULED);
        taskInfoLogs.setVersion(1);
        int insert = taskinfoLogsMapper.insert(taskInfoLogs);
        //为1则是true
        return insert == 1;
    }

    /**
     * 消费任务
     *
     * @param taskType 任务类型
     * @param priority 任务优先级
     * @return
     */
    @Override
    @Transactional
    public Task pullFromRedis(Integer taskType, Integer priority) {
        //1 拼接key
        String key = taskType + "_" + priority;
        //2 根据key从redis中获取json格式字符串
        String s = redisCacheService.lRightPop(ScheduleConstants.TOPIC + key);
        //3 解析字符串
        Task task = JSON.parseObject(s, Task.class);
        //4 TODO 要删除任务表，并且更新日志表的状态
        updateDB(task);
        return task;
    }


    /**
     * 抽取的方法
     *
     * @param task
     */
    private void updateDB(Task task) {
        taskinfoMapper.deleteById(task.getTaskId());
        TaskinfoLogs taskInfoLogs = taskinfoLogsMapper.selectById(task.getTaskId());
        if (taskInfoLogs != null) {
            //修改状态，版本自动加一
            taskInfoLogs.setStatus(ScheduleConstants.EXECUTED);
            taskinfoLogsMapper.updateById(taskInfoLogs);
        }
    }

    /**
     * 定时每分钟从zset把任务转换到list中
     */
    @Override
    @Scheduled(cron = "0/20 * * * * ? ") //每20秒钟拉取一次
    public void refreshZsetToList() {
        //获取redis分布式锁,锁30秒
        String token = redisCacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
        if (token != null) {
            //记录开始的时间
            log.info("定时任务开始拉取 {}", new Date());
            //1 从缓存中获取所有未来数据的key的集合
            Set<String> futureKeys = redisCacheService.scan(ScheduleConstants.FUTURE + "*");
            //2 遍历集合，获取每一个key
            for (String futureKey : futureKeys) {
                //3 根据futureKey获取当前需要消费的任务数据,即设置的执行时间小于等于当前时间的任务
                Set<String> tasks = redisCacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
                if (tasks != null) {
                    //4.1 生成list中所需要的key
                    String suffix = futureKey.split(ScheduleConstants.FUTURE)[1];
                    String topicKey = ScheduleConstants.TOPIC + suffix;
                    //4 将zset中任务转移到list中
                    redisCacheService.refreshWithPipeline(futureKey, topicKey, tasks);
                    log.info("成功的将" + futureKey + "下的当前需要执行的任务数据刷新到" + topicKey + "下");
                }
            }
        }
    }

    @Override
    @Scheduled(cron = "0 */5 * * * ?") //每五分钟从数据库拉取一次
    @PostConstruct //该注解的目的是为了保证在服务器重启的时候第一时间进行加载该方法，然后才是执行定时任务操作
    public void refreshDataBaseToRedis() {
        //1 清理所有的缓存，可以清理是因为只要没被消费，那么就可以从数据库再次获取，消费了数据库就删除了
        //1.1 查出所有存储未来数据的key
        Set<String> futureKeys = redisCacheService.scan(ScheduleConstants.FUTURE + "*");
        redisCacheService.delete(futureKeys);
        //1.2 查出所有存储在list中的key
        Set<String> topicKeys = redisCacheService.scan(ScheduleConstants.TOPIC + "*");
        redisCacheService.delete(topicKeys);

        //2 从任务表中查出执行时间小于等于当前时间加五分钟的任务
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE, 5);
        List<Taskinfo> taskInfos = taskinfoMapper.selectList(Wrappers.<Taskinfo>lambdaQuery().lt(Taskinfo::getExecuteTime, calendar.getTime()));

        //3 遍历任务表的数据，将属性拷贝到Task实例中，存入redis缓存中
        for (Taskinfo taskInfo : taskInfos) {
            Task task = new Task();
            BeanUtils.copyProperties(taskInfo, task);
            task.setExecuteTime(taskInfo.getExecuteTime().getTime());
            addTaskToRedis(task);
        }
    }


}
